package com.qst.order.service.Impl;


import com.qst.order.domain.DTO.QueueMessage;
import com.qst.order.domain.QueueTrend;
import com.qst.order.mapper.QueueTrendMapper;
import com.qst.order.service.RabbitMQMonitorService;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * RabbitMQ 队列监控服务
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class RabbitMQMonitorServiceImpl implements RabbitMQMonitorService {

    private final RabbitTemplate rabbitTemplate;
    private final QueueTrendMapper queueTrendMapper;

    /** 要监控的队列列表（从配置文件读取） */
    @Value("${rabbitmq.monitor.queues}")
    private List<String> monitorQueues;

    /**
     * 采集队列消息数（定时任务调用）
     */
    @Transactional(rollbackFor = Exception.class)
    public void collectQueueMessageCount() {
        LocalDateTime now = LocalDateTime.now();
        List<QueueTrend> trendList = new ArrayList<>();

        // 遍历所有监控队列，采集消息数
        for (String queueName : monitorQueues) {
            try {
                // 通过RabbitTemplate执行原生Channel操作
                Integer messageCount = rabbitTemplate.execute(channel -> {
                    // 被动声明队列（确认队列存在，不存在则抛异常）
                    channel.queueDeclarePassive(queueName);
                    // 获取队列当前消息数
                    return (int) channel.messageCount(queueName);
                });

                // 封装趋势数据
                QueueTrend trend = new QueueTrend();
                trend.setQueueName(queueName);
                trend.setMessageCount(messageCount);
                trend.setCollectTime(now);
                trendList.add(trend);
            } catch (Exception e) {
                // 单个队列采集失败不影响其他队列
                System.err.println("采集队列【" + queueName + "】消息数失败：" + e.getMessage());
            }
        }

        // 批量插入采集数据（MyBatis-Plus批量插入）
        if (!trendList.isEmpty()) {
            queueTrendMapper.insert(trendList);
        }
    }

    /**
     * 获取队列消息数趋势（时间戳列表）
     */
    public List<String> getQueueTrendTimestamps(String queueName, int timeRange) {
        LocalDateTime endTime = LocalDateTime.now();
        LocalDateTime startTime = endTime.minus(timeRange, ChronoUnit.MINUTES);
        
        List<QueueTrend> trendList = queueTrendMapper.selectByQueueNameAndTimeRange(
                queueName, startTime, endTime
        );

        // 转换时间格式为字符串（前端可直接解析）
        return trendList.stream()
                .map(trend -> trend.getCollectTime().toString())
                .collect(Collectors.toList());
    }

    /**
     * 获取队列消息数趋势（消息数列表）
     */
    public List<Integer> getQueueTrendMessageCounts(String queueName, int timeRange) {
        LocalDateTime endTime = LocalDateTime.now();
        LocalDateTime startTime = endTime.minus(timeRange, ChronoUnit.MINUTES);
        
        List<QueueTrend> trendList = queueTrendMapper.selectByQueueNameAndTimeRange(
                queueName, startTime, endTime
        );

        return trendList.stream()
                .map(QueueTrend::getMessageCount)
                .collect(Collectors.toList());
    }

    /**
     * 获取队列消息列表（前N条，不消费消息，仅查看）
     */
    public List<QueueMessage> getQueueMessages(String queueName, int pageSize) {
        List<QueueMessage> messageList = new ArrayList<>();

        // 校验队列是否在监控列表
        if (!monitorQueues.contains(queueName)) {
            throw new RuntimeException("队列【" + queueName + "】不在监控列表，无法查看消息");
        }

        try {
            for (int i = 0; i < pageSize; i++) {
                // 接收消息（autoAck=false，手动确认，避免消息被消费）
                Message message = rabbitTemplate.receive(queueName, 5000); // 1秒超时
                if (message == null) break;

                MessageProperties props = message.getMessageProperties();
                QueueMessage queueMsg = new QueueMessage();

                // 封装消息信息
                queueMsg.setMessageId(props.getMessageId() != null ? props.getMessageId() : "未知");
                queueMsg.setContent(new String(message.getBody()));
                queueMsg.setPriority(props.getPriority() != null ? props.getPriority() : 0);
                // 简化：入队时间 = 当前时间 - 消息存活时长（实际需RabbitMQ Management API）
                queueMsg.setEnqueueTime(null);
                // 处理过期时间
                if (props.getExpiration() != null) {
                    try {
                        long expiration = Long.parseLong(props.getExpiration());
                        queueMsg.setExpirationTime(LocalDateTime.now().minusNanos(expiration));
                    }catch (NumberFormatException e){
                        log.error("过期时间格式错误：" + props.getExpiration());
                        queueMsg.setExpirationTime(null);
                    }

                }
                messageList.add(queueMsg);
                // 将消息放回队列（确保业务消费不受影响）
                rabbitTemplate.send(queueName, message);
            }
        } catch (Exception e) {
            throw new RuntimeException("获取队列【" + queueName + "】消息失败：" + e.getMessage());
        }

        return messageList;
    }

    /**
     * 清空队列
     */
    public void clearQueue(String queueName) {
        // 校验队列是否在监控列表
        if (!monitorQueues.contains(queueName)) {
            throw new RuntimeException("队列【" + queueName + "】不在监控列表，禁止清空");
        }

        try {
            // 执行队列清空操作
            rabbitTemplate.execute(channel -> {
                // 被动声明队列（确认队列存在）
                channel.queueDeclarePassive(queueName);
                // 清空队列（purgeQueue：删除所有未消费消息）
                channel.queuePurge(queueName);
                return true;
            });
        } catch (Exception e) {
            throw new RuntimeException("清空队列【" + queueName + "】失败：" + e.getMessage());
        }
    }

    /**
     * 获取所有监控队列的基础信息
     */
    public List<QueueBasicInfo> getQueueBasicInfos() {
        List<QueueBasicInfo> infoList = new ArrayList<>();

        for (String queueName : monitorQueues) {
            try {
                // 获取队列基础信息（消息数、消费者数等）
                QueueBasicInfo info = rabbitTemplate.execute(channel -> {
                    channel.queueDeclarePassive(queueName);
                    long messageCount = channel.messageCount(queueName);
                    long consumerCount = channel.consumerCount(queueName);

                    QueueBasicInfo queueInfo = new QueueBasicInfo();
                    queueInfo.setName(queueName);
                    queueInfo.setMessageCount((int) messageCount);
                    queueInfo.setConsumerCount((int) consumerCount);
                    queueInfo.setStatus("running");
                    // 简化：内存占用通过消息数估算（实际需RabbitMQ Management API）
                    queueInfo.setMemoryUsage(messageCount * 10 + " KB");
                    return queueInfo;
                });
                infoList.add(info);
            } catch (Exception e) {
                // 队列异常时标记状态
                QueueBasicInfo errorInfo = new QueueBasicInfo();
                errorInfo.setName(queueName);
                errorInfo.setMessageCount(-1);
                errorInfo.setConsumerCount(-1);
                errorInfo.setStatus("error");
                errorInfo.setMemoryUsage("异常");
                infoList.add(errorInfo);
            }
        }

        return infoList;
    }

    /**
     * 队列基础信息DTO
     */
    @lombok.Data
    public static class QueueBasicInfo {
        private String name; // 队列名称
        private Integer messageCount; // 消息数
        private Integer consumerCount; // 消费者数
        private String status; // 状态（running/error）
        private String memoryUsage; // 内存占用
    }
}